package cn.wangjie.spark.kafka.batch

import org.apache.spark.sql.{DataFrame, Dataset, SaveMode, SparkSession}

/**
 * 依据偏移量范围消费Kafka中各个分区数据，进行处理，存储外部系统
 */
object SparkBatchKafka {

  def main(args: Array[String]): Unit = {
		
		// 创建SparkSession实例对象
		val spark: SparkSession = SparkSession
			.builder()
			.appName(this.getClass.getSimpleName.stripSuffix("$"))
			.master("local[2]")
			// 设置SparkSQL Shuffle时分区数目
			.config("spark.sql.shuffle.partitions", "2")
			.getOrCreate()
		// 导入隐式转换和函数库
		import spark.implicits._
		import org.apache.spark.sql.functions._
		
		// TODO: 指定偏移量，从Kafka批量加载数据
		val kafkaDF: DataFrame = spark
			.read
			.format("kafka")
			.option("kafka.bootstrap.servers", "node1.itcast.cn:9092")
			.option("subscribe", "stationTopic")
			.option("startingOffsets", "earliest") // 从最小偏移量开始读取
			.option("endingOffsets", "latest") // 读取到最大偏移量
			.load()
		val etlDF: Dataset[(String, String)] = kafkaDF
			.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
			.as[(String, String)]
		
		println(s"is Streaming DataFrame：${kafkaDF.isStreaming}")
		
		// 1. 打印数据至控制台
		etlDF.show(10, truncate = false)
		
		// 2. 保存基站日志数据至JSON文件
		etlDF.coalesce(1).write.mode(SaveMode.Overwrite).json("datas/station/")
		
		// 应用结束，关闭资源
		spark.stop()
	}
	
}
